package framework;

import java.io.IOException;
import java.util.Hashtable;

/**
 * This abstract implements a Data Sink filter. All the logic of the data sink
 * is implemented here, except the write method. The write method is implemented
 * in the specific classes that extends this one.
 */
public abstract class DataSink extends FilterFramework {

	/**
	 * Constructor
	 */
	public DataSink() {
		// creates a filter with 1 input and 0 outputs
		super(1, 0);
	}

	/**
	 * 
	 * This method reads the id of a measurement from the input pipe
	 * 
	 * @return The measurement id read
	 * @throws EndOfStreamException
	 *             if the input pipe is closed or no more data in input pipe
	 */
	protected int ReadInt() throws EndOfStreamException {
		return super.ReadInt(0);
	}

	/**
	 * 
	 * This method reads the value of a measurement from the input pipe
	 * 
	 * @return The measurement value read
	 * @throws EndOfStreamException
	 *             if the input pipe is closed or no more data in input pipe
	 */
	protected long ReadLong() throws EndOfStreamException {
		return super.ReadLong(0);
	}

	/**
	 * Abstract method that should implement the writing of a frame. Delegates
	 * the implementation to the specific classes
	 * 
	 * @param hash
	 *            The mapping between each measurement id and its value.
	 *            Represents one frame to be written.
	 * 
	 * @throws java.io.IOException
	 *             thrown if some exception while writing the frame happens
	 */
	protected abstract void write(Hashtable<Integer, Long> hash)
			throws IOException;

	/**
	 * Abstract method that should implement any code related without finishing
	 * the output writing. For instance, for a concrete DataSink implementation
	 * that writes to a file, this method should close the file handle.
	 * 
	 * @throws java.io.IOException
	 *             if an error occurs finishing the writter
	 */
	protected abstract void finish();

	/**
	 * 
	 * This method connects one filter to the output port of this filter
	 * 
	 * @param Filter
	 *            The filter that we want to connect to this filter
	 * @param OutputPortIndex
	 *            The pipe port index where we want to connect the filter.
	 */
	public void Connect(FilterFramework Filter, int OutputPortIndex) {
		super.Connect(Filter, 0, OutputPortIndex);
	}

	/**
	 * The main method of the filter. Starts when the thread is launched
	 */
	@Override
	public void run() {
		try {
			while (this.ReadInt() != 0) {
				this.ReadLong();
			}

			while (true) {

				// read time
				long measure = this.ReadLong();

				// create new hash for this new frame
				Hashtable<Integer, Long> hash = new Hashtable<Integer, Long>();

				// put the value for the time in the hash
				hash.put(Constants.TIME_ID, measure);

				try {
					for (int id = this.ReadInt(); id != 0; id = this.ReadInt()) {
						measure = this.ReadLong();
						// put the value read in the hash
						hash.put(id, measure);
					}
				} catch (EndOfStreamException e) {
					// do nothing
				}

				// writes the hash to the sink
				this.write(hash);

			} // while
		} catch (EndOfStreamException e) {
			ClosePorts();
			this.finish();
			System.out.print("\n" + this.getName() + "::Sink Exiting;");
		} // catch
		catch (Exception e) {
			ClosePorts();
			this.finish();
			System.out.print("\n" + this.getName() + "::ERROR::Sink Exiting;"
					+ e.getMessage());
		} // catch
	}
}
